package cn.doitedu.demo5;

import cn.doitedu.demo1.UserEvent_1;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

// 年龄>25岁,且 性别 = male 的用户，规则生效后，做过 C行为3次以上，当他发生 K 行为，则命中规则
// 年龄>35岁,且 性别 = female 的用户，规则生效后，做过 X行为3次以上，当他发生 P 行为，则命中规则

public class Rule3_Calculator implements RuleCalculator {

    ValueState<Integer> cntState;
    Table userProfileTable;
    JSONObject message = new JSONObject();

    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {

        // 构建hbase客户端连接，用于查询用户的画像标签
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection connection = ConnectionFactory.createConnection(hbaseConf);
        userProfileTable = connection.getTable(TableName.valueOf("user_profile"));


        // 申请一个状态，用于C事件计数
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cnt_state", Integer.class));

    }

    @Override
    public void calc(UserEvent_1 bean, Collector<String> out) throws IOException {
        // 动态画像统计
        if (bean.getEvent_id().equals("C")) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);
        }


        // 判断用户的本行为，是否是本规则要求的行为：X
        if (bean.getEvent_id().equals("K")) {


            // 先判断动态画像条件是否满足规则要求
            if (cntState.value() != null && cntState.value() >= 3) {

                // 再查询该用户规则所要求的静态画像标签
                Get get = new Get(Bytes.toBytes(bean.getUid()));
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));

                Result result = userProfileTable.get(get);

                byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                String ageStr = Bytes.toString(ageBytes);
                int age = Integer.parseInt(ageStr);

                byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                String gender = Bytes.toString(genderBytes);


                // 判断用户的画像标签是否满足规则的要求
                if (age > 25 && "male".equals(gender)) {
                    // 如果满足规则要求，则推送规则命中的结果消息
                    message.put("user_id", bean.getUid());
                    message.put("rule_id", "rule-001");
                    message.put("hit_time", bean.getTimestamp());


                    out.collect(message.toJSONString());
                }
            }

        }


    }
}
